package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.Arrays;

@Configuration
@Slf4j
public class KafkaStreamCustomerListener {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //创建kstream对象，同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        stream
                //转换流类型 key:文章id value:UpdateArticleMess
                .map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(String key, String value) {

                        UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                        return new KeyValue<String, String>(mess.getArticleId().toString(), JSON.toJSONString(mess));
                    }
                })
                //按照文章id分组
                .groupBy(new KeyValueMapper<String, String, String>() {
                    @Override
                    public String apply(String key, String value) {
                        return key;
                    }
                })
                //每隔10s统计文章
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //统计数据
                .aggregate(
                        //初始化统计结果
                        new Initializer<String>() {
                            @Override
                            public String apply() {
                                ArticleVisitStreamMess streamMess = new ArticleVisitStreamMess();
                                return JSON.toJSONString(streamMess);
                            }
                        },
                        //计算统计结果
                        new Aggregator<String, String, String>() {
                            @Override
                            public String apply(String key, String value, String aggregate) {
                                //topic发送的原始消息
                                UpdateArticleMess updateArticleMess =
                                        JSON.parseObject(value, UpdateArticleMess.class);
                                Integer add = updateArticleMess.getAdd();//收藏 评论 点赞 阅读量

                                //计算统计结果
                                ArticleVisitStreamMess articleVisitStreamMess =
                                        JSON.parseObject(aggregate, ArticleVisitStreamMess.class);
                                articleVisitStreamMess.setArticleId(Long.valueOf(key));
                                UpdateArticleMess.UpdateArticleType type = updateArticleMess.getType();
                                switch (type) {
                                    case LIKES:
                                        int likes = articleVisitStreamMess.getLike();
                                        articleVisitStreamMess.setLike(likes + add);
                                        break;
                                    case VIEWS:
                                        int views = articleVisitStreamMess.getView();
                                        articleVisitStreamMess.setView(views + add);
                                        break;
                                    case COMMENT:
                                        int comments = articleVisitStreamMess.getComment();
                                        articleVisitStreamMess.setComment(comments + add);
                                        break;
                                    case COLLECTION:
                                        int collects = articleVisitStreamMess.getCollect();
                                        articleVisitStreamMess.setCollect(collects + add);
                                }
                                return JSON.toJSONString(articleVisitStreamMess);
                            }
                        })
                .toStream()
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, String value) {
                        return new KeyValue<String, String>(key.key(), value);
                    }
                })
                //发给下一个topic处理
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return stream;
    }
}